package org.qpcr.community.device.message;

import com.alibaba.fastjson.JSONObject;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.Data;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.qpcr.community.device.entity.DeviceInstanceEntity;
import org.qpcr.community.device.enums.DeviceState;
import org.qpcr.community.device.message.property.CacheProperties;
import org.qpcr.community.device.service.LocalDeviceInstanceService;
import org.qpcr.community.gateway.external.SubscribeRequest;
import org.qpcr.community.gateway.external.SubscriptionProvider;
import org.qpcr.core.device.DeviceOperator;
import org.qpcr.core.device.DeviceRegistry;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

import reactor.core.publisher.Mono;

@Component
@RequiredArgsConstructor
@Slf4j
public class KeepOnlineSubscriptionProvider implements SubscriptionProvider {

    private final DeviceRegistry registry;
    private final LocalDeviceInstanceService deviceInstanceService;
    private Cache<String, CacheObject> mapDevices;
    private final CacheProperties cacheProperties;
    private boolean isInit = false;

    @PostConstruct
    public void init() throws InterruptedException {
        this.mapDevices = Caffeine
                .newBuilder()
                .maximumSize(cacheProperties.getMaximumSize())
                //.expireAfterWrite(cacheProperties.getExpireAfterWrite(), TimeUnit.SECONDS)
                .build();
    }

    /**
     * 初始化设备到缓存
     */
    @Scheduled(cron = "0 */5 * * * ?")
    private void initDeviceToCache() {
        QueryParamEntity param =new QueryParamEntity();
        param.noPaging().includes("id", "state", "logType");
        deviceInstanceService.query(param)
                .collectList()
                .flatMap(list -> {
                    isInit = true;
                    list.forEach(deviceInstanceEntity -> {
                        CacheObject cacheObject = mapDevices.getIfPresent(deviceInstanceEntity.getId());
                        if(cacheObject==null){
                            cacheObject = new CacheObject();
                        }
//                        if (mapDevices.getIfPresent(deviceInstanceEntity.getId()) == null) {
                            cacheObject.setTime(LocalDateTime.now());
                            cacheObject.setDeviceState(deviceInstanceEntity.getState());
                            cacheObject.setLogType(deviceInstanceEntity.getLogType());
                            mapDevices.put(deviceInstanceEntity.getId(), cacheObject);
//                        }
                    });
                    return null;
                })
                .subscribe();
    }

    @Data
    class CacheObject {
        private LocalDateTime time;
        private DeviceState deviceState;
        private Integer logType;
    }

    @Override
    public String id() {
        return "keep-device-online";
    }

    @Override
    public String name() {
        return "保持设备在线";
    }

    @Override
    public String[] getTopicPattern() {
        return new String[]{
                "/realtime/keep-device-online"
        };
    }




    @Override
    public Flux<Map<String, Object>> subscribe(SubscribeRequest request) {
        String deviceId = request.get("deviceId")
                .map(String.class::cast)
                .orElseThrow(() -> new IllegalArgumentException("deviceId不能为空"));
        String productId = request.get("productId")
                                 .map(String.class::cast)
                                 .orElseThrow(() -> new IllegalArgumentException("productId不能为空"));

        CacheObject cacheObjectOld = mapDevices.getIfPresent(deviceId);
//        if(cacheObjectOld.getDeviceState()==DeviceState.offline){
            //如果缓存是离线，则调用上线方法
            WebClient webClient = WebClient.create("http://127.0.0.1:1884");
            String uri="/"+productId+"/"+deviceId+"/online";
            Mono<String> response = webClient.post()
                                             .uri(uri)
                                             .bodyValue("{}").header("Authorization","Bearer admin")
                                             .retrieve()
                                             .bodyToMono(String.class)
                                             .log();
        log.info("自动发送online上线请求");

//            response.subscribe(System.out::println);
        response.subscribe(
            body -> System.out.println("----------Success: " + body),
            error -> System.err.println("---------Error: " + error.getMessage())
        );
//        }


        Map<String, Object> map = new HashMap<>();
        map.put("deviceId", deviceId);
        CacheObject cacheObject = new CacheObject();
        cacheObject.setTime(LocalDateTime.now());
        cacheObject.setDeviceState(DeviceState.online);
        mapDevices.put(deviceId, cacheObject);
        // todo 后面删掉
        log.debug("在线ping:{}", deviceId);
        return Flux.just(map);

    }

    @Scheduled(cron = "*/5 * * * * ?")
    public void offine() {
        // 在PostConstruct不能初始化反应式，在这里初始化
        if (Boolean.FALSE.equals(isInit)) {
            this.initDeviceToCache();
        }
        ConcurrentMap<String, CacheObject> map = mapDevices.asMap();
        for(String key : map.keySet()) {
            CacheObject cacheObject = mapDevices.getIfPresent(key);
            // 计算两个日期时间之间的差异
            Duration duration = Duration.between(cacheObject.getTime(), LocalDateTime.now());
            duration.getSeconds();
            // 超过5秒就下线
            if (cacheObject.getDeviceState() == DeviceState.online && duration.getSeconds() > cacheProperties.getBetweenAt()) {

                if(cacheObject.getLogType()!=null && cacheObject.getLogType()==1){
                    deviceInstanceService.DelDeviceLogState(key).doOnSuccess(unused -> System.out.println("Update executed successfully"))
                                         .doOnError(error -> System.err.println("Error occurred during update: " + error.getMessage()))
                                         .subscribe();
                }



                registry
                        .getDevice(key)
                        .flatMapMany(DeviceOperator::disconnect)
                        .singleOrEmpty()
                        .subscribe();

                cacheObject.setDeviceState(DeviceState.offline);
                mapDevices.put(key, cacheObject);

//                 map.remove(key);
                log.info("设备{}长时间接收不到新消息，下线。", key);

            }
        }
    }
}
